package com.test.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KafkaJob {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.准备数据

//        final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList(
//                new TopicPartition("topic-a", 0),    // Partition 0 of topic "topic-a"
//                new TopicPartition("topic-b", 5)));  // Partition 5 of topic "topic-b"
//        KafkaSource.builder().setPartitions(partitionSet);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("test")
                .setGroupId("flink")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> stream = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks(),
                "kafka source"
        );

        //3.设置flink 任务
        stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                return "flink : " + s;
            }
        }).print();

        // Execute program, beginning computation.
        env.execute("flink and kafka connection");
    }
}
